package chapter3.eg2;

import java.util.Properties;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.junit.Test;

public class CustomerSerializerApp {
  @Test
  public void testCustomSerializer() {
    Properties properties = new Properties();
    properties.put("bootstrap.servers", "localhost:9092");
    properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    properties.put("value.serializer", "chapter3.eg2.CustomerSerializer");
    KafkaProducer<String, Customer> producer = new KafkaProducer<>(properties);

    try {
      while (true) {
        Customer customer = new Customer(1, "zhangsan");
        ProducerRecord<String, Customer> record = new ProducerRecord<>("customerCountries", "1",
            customer);
        producer.send(record);
        System.out.println("send customer record: " + customer);
        TimeUnit.SECONDS.sleep(1);
      }
    } catch (Exception e) {
      e.printStackTrace();
    }
  }

}
